package com.example.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.example.configs.SysTaskConfig;
import com.example.entity.SysTask;
import com.example.mapper.SysTaskMapper;
import com.example.server.SysServer;
import com.example.service.SysTaskDispatcherService;
import com.example.service.SysTaskService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
public class SysTaskServiceImpl extends ServiceImpl<SysTaskMapper, SysTask> implements SysTaskService, InitializingBean, DisposableBean {

    Logger logger = LoggerFactory.getLogger(getClass());

    private LinkedBlockingQueue<SysTask> fastQueue = new LinkedBlockingQueue<>();

    private DelayQueue<SysTask> errorQueue = new DelayQueue<>();

    private ThreadPoolExecutor threadPoolExecutor = null;

    @Resource
    private SysTaskDispatcherService sysTaskDispatcherService;

    @Resource
    private SysServer sysServer;

    @Resource
    private SysTaskConfig sysTaskConfig;

    private Thread fastThread;

    private Thread errorThread;

    @Override
    public void afterPropertiesSet() throws Exception {
        this.init();
    }

    @Override
    public void destroy() throws Exception {
        this.stop();
    }

    @Override
    public void init() {
        this.initTreadPool();
        this.initQueue();
        this.startWorkThread();
    }

    @Override
    public void initTreadPool() {
        this.threadPoolExecutor = new ThreadPoolExecutor(sysTaskConfig.getCorePoolSize(), sysTaskConfig.getMaximumPoolSize(), sysTaskConfig.getKeepAliveTime(), TimeUnit.SECONDS, new LinkedBlockingQueue<>());
    }

    @Override
    public void initQueue() {
        this.refreshQueue(0L);
    }

    @Override
    public void startWorkThread() {
        this.fastThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        SysTask sysTask = fastQueue.take();
                        threadPoolExecutor.execute(new Runnable() {
                            @Override
                            public void run() {
                                sysTaskDispatcherService.updateAndHandler(sysTask);
                            }
                        });
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        this.fastThread.setName("fastThread");
        this.fastThread.setDaemon(true);
        this.fastThread.start();
        this.errorThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        SysTask sysTask = errorQueue.take();
                        sysTaskDispatcherService.updateAndHandler(sysTask);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        this.errorThread.setName("errorThread");
        this.errorThread.setDaemon(true);
        this.errorThread.start();
    }

    @Override
    public void stop() {
        this.stopTreadPool();
        this.stopWorkThread();
    }

    @Override
    public void stopWorkThread() {
        if (this.fastThread.isAlive()) {
            this.fastThread.interrupt();
        }
        if (this.errorThread.isAlive()) {
            this.errorThread.interrupt();
        }
    }

    @Override
    public void stopTreadPool() {
        if (!this.threadPoolExecutor.isShutdown()) {
            this.threadPoolExecutor.shutdown();
        }
    }

    @Override
    public void refreshQueue(Long syncTime) {
        String serverKey = this.sysServer.getServerKey();
        List<SysTask> fastList = getBaseMapper().refreshFastQueue(serverKey, syncTime);
        List<SysTask> errorList = getBaseMapper().refreshErrorQueue(serverKey, syncTime, FAILURES_COUNT);
        if (!fastList.isEmpty()) {
            boolean flag = this.fastQueue.addAll(fastList);
            if (!flag) {
                logger.warn("fastQueue已达到极限，无法加入对列: {}", fastList.size());
            }
        }
        if (!errorList.isEmpty()) {
            errorList.stream().map((item) -> {
                item.setRunTime(System.nanoTime());
                return item;
            }).collect(Collectors.toList());
            boolean flag = this.errorQueue.addAll(errorList);
            if (!flag) {
                logger.warn("errorQueue已达到极限，无法加入对列: {}", errorList.size());
            }
        }
    }


    /**
     * 单个任务分发
     *
     * @param sysTask
     */
    @Override
    public void dispatcher(SysTask sysTask) {
        String serverKey = this.sysServer.getServerKey();
        sysTask.setServerKey(serverKey);
        sysTask.setJoinTime(System.currentTimeMillis());
        this.fastQueue.add(sysTask);
        this.save(sysTask);
    }

    @Override
    public void dispatcherToErrorQueue(SysTask sysTask) {
        this.errorQueue.add(sysTask);
    }

    /**
     * 针对大批量分发
     *
     * @param sysTasks
     */
    @Override
    public void updateAndDispatcherListToQueue(List<SysTask> sysTasks) {
        Long nowTime = System.currentTimeMillis();
        List<String> serverKeyList = this.sysServer.getServerKeyList();
        if (serverKeyList != null && serverKeyList.size() > 0) {
            int count = sysTasks.size() / serverKeyList.size();
            for (int kk = 1; kk <= serverKeyList.size(); kk++) {
                int start = (kk - 1) * count;
                int end = kk * count;
                if (kk == serverKeyList.size()) {
                    end = sysTasks.size();
                }
                final String sk = serverKeyList.get(kk - 1);
                List<SysTask> targetList = sysTasks.subList(start, end);
                targetList.stream().map((item) -> {
                    Long now = nowTime + 1000;
                    item.setJoinTime(now);
                    item.setServerKey(sk);
                    return item;
                }).collect(Collectors.toList());
                this.saveOrUpdateBatch(targetList);
            }
            this.sysServer.notifyAllServersToRefresh(nowTime);
        }
    }

    /**
     * 查询还没有完成的任务并转移到另一个节点.
     *
     * @param serverKey
     * @return
     */
    @Override
    public List<SysTask> findTaskByOffline(String serverKey) {
        return getBaseMapper().findTaskByOffline(serverKey, FAILURES_COUNT);
    }

}
